﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="MsSqlBulkCopyConsumer.cs" company="DHGMS Solutions">
//   Copyright 2004-2014 DHGMS Solutions.
//      
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//      
//   http://www.apache.org/licenses/LICENSE-2.0
//      
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.
// </copyright>
// <summary>
//   TODO: Update summary.
// </summary>
// --------------------------------------------------------------------------------------------------------------------

namespace Dhgms.TplHelper.Model.DataReaderConsumer
{
    using System.Threading.Tasks;

    using System;
    using System.Data;
    using System.Data.SqlClient;
    using System.Text;

    using Dhgms.DataManager.Model.Helper.Database;
    using Dhgms.TplHelper.Model.Info;

    /// <summary>
    /// TODO: Update summary.
    /// </summary>
    public class MsSqlBulkCopyTarget : IDataReaderConsumer<long>
    {
        /// <summary>The number of rows copied.</summary>
        private long rowsCopied;

        /// <summary>
        /// the delete action to be taken on the target table
        /// </summary>
        private readonly DestinationTableDeleteAction targetTableDeleteAction;

        /// <summary>Initializes a new instance of the <see cref="MsSqlBulkCopyTarget"/> class.</summary>
        /// <param name="destinationConnectionString">The destination connection string.</param>
        /// <param name="destinationDatabase">The destination database.</param>
        /// <param name="destinationSchema">The destination schema.</param>
        /// <param name="destinationTable">The destination table.</param>
        /// <param name="targetTableDeleteAction">
        /// the delete action to be taken on the target table
        /// </param>
        public MsSqlBulkCopyTarget(
            string destinationConnectionString,
            string destinationDatabase,
            string destinationSchema,
            string destinationTable,
            DestinationTableDeleteAction targetTableDeleteAction)
        {
            if (string.IsNullOrWhiteSpace(destinationConnectionString))
            {
                throw new ArgumentNullException(destinationConnectionString);
            }

            if (string.IsNullOrWhiteSpace(destinationDatabase))
            {
                throw new ArgumentNullException(destinationDatabase);
            }

            if (string.IsNullOrWhiteSpace(destinationSchema))
            {
                throw new ArgumentNullException(destinationSchema);
            }

            if (string.IsNullOrWhiteSpace(destinationTable))
            {
                throw new ArgumentNullException(destinationTable);
            }

            this.DestinationConnectionString = destinationConnectionString;
            this.DestinationDatabase = destinationDatabase;
            this.DestinationSchema = destinationSchema;
            this.DestinationTable = destinationTable;
            this.targetTableDeleteAction = targetTableDeleteAction;
        }

        /// <summary>
        /// Destination Database Connection String
        /// </summary>
        protected string DestinationConnectionString { get; private set; }

        /// <summary>
        /// the name of the destination database
        /// </summary>
        protected string DestinationDatabase { get; private set; }

        /// <summary>
        /// the schema the destination table sits in
        /// </summary>
        protected string DestinationSchema { get; private set; }

        /// <summary>
        /// the table the data is going to end up in
        /// </summary>
        protected string DestinationTable { get; private set; }

        /// <summary>Processes data reader.</summary>
        /// <param name="dataReader">The data reader.</param>
        /// <returns>The number of records processed</returns>
        public async Task<DataRetrievalProcessorResult<long>> DataReaderAction(IDataReader dataReader)
        {
            var databaseHelper = new MSSql10();
            var dataTable = new DataTable();
            dataTable.Load(dataReader);
            switch (this.targetTableDeleteAction)
            {
                case DestinationTableDeleteAction.Truncate:
                    databaseHelper.TruncateTable(
                        this.DestinationConnectionString,
                        this.DestinationDatabase,
                        this.DestinationSchema,
                        this.DestinationTable);
                    break;
                case DestinationTableDeleteAction.WipeAndReseed:
                    databaseHelper.WipeAndReseed(
                        this.DestinationConnectionString,
                        this.DestinationDatabase,
                        this.DestinationSchema,
                        this.DestinationTable);
                    break;
                case DestinationTableDeleteAction.None:
                    break;
            }

            var destinationTableName = "[" + this.DestinationDatabase + "].[" + this.DestinationSchema + "].["
                                       + this.DestinationTable + "]";
            try
            {
                using (var bulkCopy = new SqlBulkCopy(this.DestinationConnectionString, SqlBulkCopyOptions.TableLock))
                {
                    bulkCopy.DestinationTableName = destinationTableName;
                    bulkCopy.BulkCopyTimeout = 0;
                    bulkCopy.NotifyAfter = 10000;
                    bulkCopy.SqlRowsCopied += this.BulkCopySqlRowsCopied;

                    await bulkCopy.WriteToServerAsync(dataTable);
                    bulkCopy.SqlRowsCopied -= this.BulkCopySqlRowsCopied;
                }
            }
            catch (Exception exception)
            {
                // loop through all inner exceptions to see if any relate to a constraint failure
                bool dataExceptionFound = false;
                var tmpException = exception;

                while (tmpException != null)
                {
                    if (tmpException is SqlException && tmpException.Message.Contains("constraint"))
                    {
                        dataExceptionFound = true;
                        break;
                    }

                    tmpException = tmpException.InnerException;
                }


                if (!dataExceptionFound)
                {
                    // call the helper method to document the errors and invalid data
                    string errorMessage = GetBulkCopyFailedData(
                        this.DestinationConnectionString,
                        destinationTableName,
                        dataTable.CreateDataReader());
                    throw new Exception(errorMessage, exception);
                }
            }

            return new DataRetrievalProcessorResult<long>(this.rowsCopied, 0);
        }

        private void BulkCopySqlRowsCopied(object sender, SqlRowsCopiedEventArgs e)
        {
            this.rowsCopied = this.rowsCopied + e.RowsCopied;
        }

        /// <summary>
        /// Build an error message with the failed records and their related exceptions.
        /// </summary>
        /// <param name="connectionString">Connection string to the destination database</param>
        /// <param name="tableName">Table name into which the data will be bulk copied.</param>
        /// <param name="dataReader">DataReader to bulk copy</param>
        /// <returns>Error message with failed constraints and invalid data rows.</returns>
        /// <remarks>
        /// Taken from http://www.codeproject.com/Articles/387465/Retrieving-failed-records-after-an-SqlBulkCopy-exc
        /// </remarks>
        private static string GetBulkCopyFailedData(string connectionString, string tableName, IDataReader dataReader)
        {
            StringBuilder errorMessage = new StringBuilder("Bulk copy failures:" + Environment.NewLine);
            SqlConnection connection = null;
            SqlTransaction transaction = null;
            SqlBulkCopy bulkCopy = null;
            DataTable tmpDataTable = new DataTable();

            try
            {
                connection = new SqlConnection(connectionString);
                connection.Open();
                transaction = connection.BeginTransaction();
                bulkCopy = new SqlBulkCopy(connection, SqlBulkCopyOptions.CheckConstraints, transaction);
                bulkCopy.DestinationTableName = tableName;

                // create a datatable with the layout of the data.
                DataTable dataSchema = dataReader.GetSchemaTable();
                foreach (DataRow row in dataSchema.Rows)
                {
                    tmpDataTable.Columns.Add(new DataColumn(row["ColumnName"].ToString(), (Type)row["DataType"]));
                }

                // create an object array to hold the data being transferred into tmpDataTable 
                //in the loop below.
                object[] values = new object[dataReader.FieldCount];

                // loop through the source data
                while (dataReader.Read())
                {
                    // clear the temp DataTable from which the single-record bulk copy will be done
                    tmpDataTable.Rows.Clear();

                    // get the data for the current source row
                    dataReader.GetValues(values);

                    // load the values into the temp DataTable
                    tmpDataTable.LoadDataRow(values, true);

                    // perform the bulk copy of the one row
                    try
                    {
                        bulkCopy.WriteToServer(tmpDataTable);
                    }
                    catch (Exception ex)
                    {
                        // an exception was raised with the bulk copy of the current row. 
                        // The row that caused the current exception is the only one in the temp 
                        // DataTable, so document it and add it to the error message.
                        DataRow faultyDataRow = tmpDataTable.Rows[0];
                        errorMessage.AppendFormat("Error: {0}{1}", ex.Message, Environment.NewLine);
                        errorMessage.AppendFormat("Row data: {0}", Environment.NewLine);
                        foreach (DataColumn column in tmpDataTable.Columns)
                        {
                            errorMessage.AppendFormat(
                                "\tColumn {0} - [{1}]{2}",
                                column.ColumnName,
                                faultyDataRow[column.ColumnName].ToString(),
                                Environment.NewLine);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to document SqlBulkCopy errors. See inner exceptions for details.", ex);
            }
            finally
            {
                if (transaction != null)
                {
                    transaction.Rollback();
                }
                if (connection.State != ConnectionState.Closed)
                {
                    connection.Close();
                }
            }
            return errorMessage.ToString();
        }

    }
}